package com.zhb.cloud.curator.utils;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorFrameworkUtils {
	/**
	 * NAME_SPACE
	 */
	private static final String NAME_SPACE = "zhb/zk";
	/**
	 * zookeeper hosts address
	 */
	private static final String CONNECT_STRING = "192.168.1.109:2181,192.168.1.109:2182,192.168.1.109:2183";
	/**
	 * session timeout in milliseconds
	 */
	public static final int ZOOKEEPER_SESSION_TIMEOUT = 10 * 1000;

	/**
	 * connection timeout in milliseconds
	 */
	public static final int ZOOKEEPER_CONNECTION_TIMEOUT = 10 * 1000;

	/**
	 * max number of times to retry
	 */
	public static final int ZOOKEEPER_RETRY_TIMES = 3;

	/**
	 * initial amount of time to wait between retries in milliseconds
	 */
	public static final int ZOOKEEPER_RETRY_INTERVAL = 1000;

	/**
	 * @Title: initCuratorFramework
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:57:47
	 * @Description: 初始化
	 * @return
	 */
	public static CuratorFramework initCuratorFramework() {
		return getCuratorFramework(CONNECT_STRING, NAME_SPACE);
	}

	/***
	 * @Title: getCuratorFramework
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:02:49
	 * @Description: 获取zk客户端
	 * @param connectString
	 * @param namespace
	 * @return
	 */
	private static CuratorFramework getCuratorFramework(String connectString, String namespace) {
		if (StringUtils.isEmpty(connectString)) {
			throw new NullPointerException("empty zookeeper connect string");
		}
		// normalize namespace
		if (!StringUtils.isBlank(namespace)) {
			namespace = namespace.replaceAll("^/+", "");
		}
		// retry policy

		RetryPolicy retryPolicy = new ExponentialBackoffRetry(ZOOKEEPER_RETRY_INTERVAL, ZOOKEEPER_RETRY_TIMES);
		// build zk client
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString(connectString)
				.sessionTimeoutMs(ZOOKEEPER_SESSION_TIMEOUT).connectionTimeoutMs(ZOOKEEPER_CONNECTION_TIMEOUT)
				.canBeReadOnly(true).retryPolicy(retryPolicy);
		if (!StringUtils.isBlank(namespace)) {
			builder.namespace(namespace);
		}
		CuratorFramework client = builder.build();
		// start zk client
		if (client.getState() != CuratorFrameworkState.STARTED) {
			client.start();
		}
		return client;
	}

}
